package com.topn;

import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.yarn.webapp.hamlet.Hamlet;

import java.io.*;
import java.net.URI;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

public class TMapper extends Mapper<LongWritable, Text, Tkey, IntWritable> {

    private Tkey mkey = new Tkey();
    private IntWritable mval = new IntWritable();

    private Map<String, String> locMap = new HashMap<>();


    // 把小表的数据加载到内存
    // 初始化方法，在map方法调用之前会把setup方法执行完毕
    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        String path = context.getCacheFiles()[0].getPath();
        Path locationPath = new Path(path);
        // 获取hdfs文件系统
        FileSystem fs = locationPath.getFileSystem(context.getConfiguration());

        // 打开文件的输入流
        FSDataInputStream in = fs.open(locationPath);

        BufferedReader br = new BufferedReader(new InputStreamReader(in));

        String line;
        while ((line = br.readLine()) != null) {
            String[] arr = line.split("\t");
            locMap.put(arr[0], arr[1]);
        }

    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        System.out.println(locMap);
        //  String str = "2019-6-1 22:22:22\t2\t36";
        Map<String, String> map = processLine(value.toString());

        mkey.setYear(Integer.valueOf(map.get(DateEnum.YEAR.getFlag())));
        mkey.setMonth(Integer.valueOf(map.get(DateEnum.MONTH.getFlag())));
        mkey.setDay(Integer.valueOf(map.get(DateEnum.DAY.getFlag())));
        mkey.setTemp(Integer.valueOf(map.get(DateEnum.TEMP.getFlag())));

        // join 小表得到的地址
        String location = map.get("location");
        mkey.setLocation(location);

        mval.set(Integer.valueOf(map.get(DateEnum.TEMP.getFlag())));
        context.write(mkey, mval);

    }


    private Map<String, String> processLine(String line) {
        try {
            //  String str = "2019-6-1 22:22:22\t2\t36";
            String[] arr = line.split("\t");
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
            Date date = sdf.parse(arr[0]);


            // 日历工具类
            Calendar calendar = Calendar.getInstance();
            calendar.setTime(date);

            int year = calendar.get(Calendar.YEAR);
            int month = calendar.get(Calendar.MONTH) + 1;
            int day = calendar.get(Calendar.DAY_OF_MONTH);
            int temp = Integer.valueOf(arr[arr.length - 1]);

            // 解析 地址的id， 从map中获取地址value
            String key = arr[1];
            String location = locMap.get(key);

            Map<String, String> result = new HashMap<>();
            result.put(DateEnum.YEAR.getFlag(), year + "");
            result.put(DateEnum.MONTH.getFlag(), month + "");
            result.put(DateEnum.DAY.getFlag(), day + "");
            result.put(DateEnum.TEMP.getFlag(), temp + "");
            result.put("location", location);
            return result;
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }


}
